package com.yl.flink.sink;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.json.JSONUtil;
import com.yl.constant.Const;
import com.yl.constant.RedisCmd;
import com.yl.entity.MultiDataEntity;
import com.yl.entity.PrimaryWarn;
import com.yl.entity.cdc.BatchIntegrity;
import com.yl.flink.processor.CalPointMapFunc;
import com.yl.flink.processor.RawPointMapFunc;
import com.yl.flink.redis.RedisCmdDesc;
import com.yl.flink.redis.RedisMapper;
import com.yl.flink.redis.RedisSinkMapper;
import com.yl.util.FlinkUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.influxdb.InfluxDBPoint;

import java.util.concurrent.TimeUnit;

/**
 * @author wlf
 * @since 2022/8/18
 */
public class SinkTool {

    /**
     * 存储原始数据
     *
     * @param rawStream 原始数据流
     * @param params    环境参数
     */
    public static void sinkRawStream(SingleOutputStreamOperator<MultiDataEntity> rawStream, ParameterTool params) {
        String flinkSink = params.get(Const.FLINK_SINK);
        if (Const.S_INFLUXDB.equals(flinkSink)) {
            // 存储到influxdb
            customInfluxDBSink(rawStream, new RawPointMapFunc(), params);
        } else if (Const.S_TDENGINE.equals(flinkSink)) {
            // 存储到tdengine
            // TODO: 2022/9/13 存储到tdengine
            rawStream
                    .addSink(new TDengineSink(params));

        } else if (Const.S_ALL.equals(flinkSink)) {
            // 存储到influxdb和tdengine
            // TODO: 2022/9/13 存储到influxdb和tdengine

        }
    }

    /**
     * 存储计算数据
     *
     * @param calStream 计算数据流
     * @param params    环境参数
     */
    public static void sinkCalStream(SingleOutputStreamOperator<MultiDataEntity> calStream, ParameterTool params) {
        String flinkSink = params.get(Const.FLINK_SINK);
        if (Const.S_INFLUXDB.equals(flinkSink)) {
            // 存储到influxdb
            customInfluxDBSink(calStream, new CalPointMapFunc(), params);
        } else if (Const.S_TDENGINE.equals(flinkSink)) {
            // 存储到tdengine
            // TODO: 2022/9/13 存储到tdengine

        } else if (Const.S_ALL.equals(flinkSink)) {
            // 存储到influxdb和tdengine
            // TODO: 2022/9/13 存储到influxdb和tdengine

        }
    }

    /**
     * 把数据写入到influxdb中
     */
    private static void customInfluxDBSink(SingleOutputStreamOperator<MultiDataEntity> stream, MapFunction<MultiDataEntity, InfluxDBPoint> mapFunction, ParameterTool params) {
        // 存储到influxdb
        stream
                // 组装计算数据InfluxDBPoint
                .map(mapFunction)
                // 使用influxdb-connector写入数据
                .addSink(new CustomInfluxSink(params));
    }

    /**
     * 把告警信息写入MySQL
     */
    public static SinkFunction<PrimaryWarn> warnMysqlSink(ParameterTool params) {
        return JdbcSink.sink(
                // 执行sql-upsert
                Const.SQL_INSERT_WARN,
                // 设置占位符与字段的对应关系
                (ps, warn) -> {
                    ps.setString(1,
                            JSONUtil.toJsonStr(MapUtil.of(Const.S_DATA, warn.getMultiDataEntity().getRawValues())));
                    ps.setString(2, JSONUtil.toJsonStr(warn.getTh()));
                    ps.setTimestamp(3, DateUtil.date(warn.getWarnTime()).toTimestamp());
                    ps.setTimestamp(4, DateUtil.date(warn.getTime()).toTimestamp());
                    ps.setInt(5, warn.getWarn().getLevel());
                    ps.setString(6, warn.getMessage());
                    ps.setInt(7, warn.getMultiDataEntity().getProject().getId());
                    ps.setInt(8, warn.getQuota().getId());
                    ps.setInt(9, warn.getMultiDataEntity().getTarget().getId());
                    ps.setInt(10, warn.getMultiDataEntity().getTargetType().getId());
                    ps.setDouble(11, warn.getDataValue());
                },
                // 执行配置
                getMysqlExecutionOptions(),
                // 连接配置
                getMysqlConnectionOptions(params)
        );
    }

    /**
     * 把包完整性信息写入MySQL
     */
    public static SinkFunction<BatchIntegrity> batchIntegrityMysqlSink(ParameterTool params) {
        return JdbcSink.sink(
                // 执行sql-upsert
                Const.SQL_UPSERT_DBI,
                // 设置占位符与字段的对应关系
                (ps, batchIntegrity) -> {
                    ps.setInt(1, batchIntegrity.getTarget_id());
                    ps.setString(2, batchIntegrity.getCj_day());
                    ps.setInt(3, batchIntegrity.getBatch_id());
                    ps.setInt(4, batchIntegrity.getPacket_sum_design());
                    ps.setInt(5, batchIntegrity.getPacket_sum_receive());
                    ps.setInt(6, batchIntegrity.getIntegrity());
                    ps.setString(7, batchIntegrity.getStart_time());
                    ps.setString(8, batchIntegrity.getEnd_time());
                    ps.setString(9, batchIntegrity.getCreate_time());
                },
                // 执行配置
                getMysqlExecutionOptions(),
                // 连接配置
                getMysqlConnectionOptions(params)
        );
    }

    /**
     * 获取mysql执行配置
     */
    private static JdbcExecutionOptions getMysqlExecutionOptions() {
        return JdbcExecutionOptions.builder()
                .withBatchSize(5)
                .withBatchIntervalMs(5000)
                .build();
    }

    /**
     * 获取mysql连接配置
     */
    private static JdbcConnectionOptions getMysqlConnectionOptions(ParameterTool params) {
        return new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withDriverName(params.get(Const.MYSQL_DRIVER))
                .withUrl(params.get(Const.MYSQL_URL))
                .withUsername(params.get(Const.MYSQL_USERNAME))
                .withPassword(params.get(Const.MYSQL_PASSWORD))
                .build();
    }

    /**
     * 把包完整性信息写入MySQL
     * 【因不能设置ttl而废弃】
     */
    public static SinkFunction<Tuple3<String, Integer, Integer>> customRedisSink(ParameterTool params) {
        return new CustomRedisSink<>(FlinkUtil.getJedisPoolConfig(params), new RedisSinkMapper());
    }
}
